package com.sfx.messagequeue.mqServer.datacenter;

import com.sfx.messagequeue.common.exception.MQException;
import com.sfx.messagequeue.mqServer.core.Binding;
import com.sfx.messagequeue.mqServer.core.Exchange;
import com.sfx.messagequeue.mqServer.core.MSGQueue;
import com.sfx.messagequeue.mqServer.core.Message;
import lombok.Data;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description: 用于统一管理内存中的数据
 * User: sfx
 * Date: 2023-08-04
 * Time: 16:46
 */
@Data
public class MemoryDataCenter {
    //交换机管理 key: exchangeName  val: Exchange
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //队列管理  key: queueName  val: MSGQueue
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    //绑定管理  key: exchangeName  key: queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //消息管理 key: messageId val:Message
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //队列与消息之间的关系 --> 表示一个队列中有哪些消息 key: queueName  val: LinkedList<Message>
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //未被确认的消息管理 --> 表示被消费者处理的消息,有哪些没有被确认
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    //插入交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("[MemoryDataCenter] 插入交换机成功 queueName=" + exchange.getName());
    }

    //删除交换机
    public void deleteExchange(Exchange exchange) {
        exchangeMap.remove(exchange.getName());
        System.out.println("[MemoryDataCenter] 删除交换机成功 queueName=" + exchange.getName());
    }

    //根据name获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    //插入队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("[MemoryDataCenter] 插入队列成功 queueName=" + queue.getName());
    }

    //删除队列
    public void deleteQueue(MSGQueue queue) {
        queueMap.remove(queue.getName());
        System.out.println("[MemoryDataCenter] 删除队列成功 queueName=" + queue.getName());
    }

    //根据name获取队列
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    //插入绑定
    public void insertBinding(Binding binding) throws MQException {
        //bindingMap为空则创建,不为空则不创建
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), (key) -> new ConcurrentHashMap<>());
        //插入的时候可能有多个请求来插入,所以就有可能出现线程安全问题
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MQException("[MemoryDataCenter:] 绑定已经存在 queueName=" + binding.getQueueName() + " exchangeName=" + binding.getExchangeName());
            }
            //不存在binding才插入绑定
            bindingMap.put(binding.getQueueName(), binding);
            System.out.println("[MemoryDataCenter] 插入绑定成功 exchangeName=" +
                    binding.getExchangeName() + " queueName=" + binding.getQueueName());
        }
    }

    //删除绑定
    public void deleteBinding(Binding binding) throws MQException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MQException("[MemoryDataCenter:] Binding不存在");
        }
        bindingsMap.remove(binding.getExchangeName());
        System.out.println("[MemoryDataCenter] 删除绑定成功 exchangeName=" +
                binding.getExchangeName() + " queueName=" + binding.getQueueName());
    }

    //根据name获取绑定
    public Binding getBinding(String queueName, String exchangeName) throws MQException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    //根据exchangeName获取所有绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    //添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryDataCenter] 添加消息成功 messageId=" +
                message.getMessageId());
    }

    //根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    //根据id删除消息
    public void deleteMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 删除消息成功 messageId=" + messageId);
    }

    //投递消息到队列
    public void sendMessage(MSGQueue queue, Message message) {
        //1.根据queueName看队列中有没有消息,如果没有则创建一个消息链表
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), (key) -> new LinkedList<>());
        //2.将消息加入到链表中
        synchronized (message) {
            messages.add(message);
        }
        //3.将message加入到消息中心中
        addMessage(message);
        System.out.println("[MemoryDataCenter] 投递消息到队列成功 messageId=" + message.getMessageId()+ " queueName " + queue.getName());
    }

    //从队列中获取消息-头删
    public Message pollMessage(String queueName) {
        //1.找到队列
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            Message message = messages.remove(0);
            System.out.println("[MemoryDataCenter] 队列获取元素成功 messageId=" + message.getMessageId() + " queueName " + queueName);
            return message;
        }
    }

    //获取指定队列中的消息个数
    public int messageSizeByQueue(String queueName) {
        //1.找到队列
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            if (messages.size() == 0) {
                return 0;
            }
            int size = messages.size();
            System.out.println("[MemoryDataCenter] 获取指定队列中的消息个数 size=" + size + " queueName " + queueName);
            return size;
        }
    }

    //添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message) {
        //根据queueName看是否有对应的消息map,如果没有则创建
        ConcurrentHashMap<String, Message> mqWaitAck = queueMessageWaitAckMap.computeIfAbsent(queueName, (k) -> new ConcurrentHashMap<>());
        //添加未确认消息
        mqWaitAck.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 添加未确认的消息成功 messageId=" + message.getMessageId() + " queueName " + queueName);
    }

    //删除未确认的消息
    public void removeMessageWaitAck(String queueName,Message message) {
        //看看有没有这个队列
        ConcurrentHashMap<String, Message> mqWaitAck = queueMessageWaitAckMap.get(queueName);
        if(mqWaitAck == null || mqWaitAck.isEmpty()) {
            return;
        }
        mqWaitAck.remove(message.getMessageId());
        System.out.println("[MemoryDataCenter] 删除未确认的消息成功 messageId=" + message.getMessageId() + " queueName " + queueName);
    }

    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> mqWaitAck = queueMessageWaitAckMap.get(queueName);
        if(mqWaitAck == null || mqWaitAck.isEmpty()) {
            return null;
        }
        return mqWaitAck.get(messageId);
    }

    /**
     * 将硬盘中的数据全部加载到内存中
     * @param diskDataCenter 操作磁盘的类
     */
    public void recovery(DiskDataCenter diskDataCenter) throws MQException, IOException, ClassNotFoundException {
        //0. 加载前,先把内存中数据结构进行清空处理
        queueMap.clear();
        messageMap.clear();
        bindingsMap.clear();
        queueMessageMap.clear();
        exchangeMap.clear();
        queueMessageWaitAckMap.clear();
        //1. 加载交换机数据 - exchangeMap
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        //2. 加载队列数据 - queueMap
        List<MSGQueue> msgQueues = diskDataCenter.selectAllQueue();
        for (MSGQueue msgQueue : msgQueues) {
            queueMap.put(msgQueue.getName(),msgQueue);
        }
        //3. 加载绑定数据 - bindingsMap
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            //key: exchangeName  key: queueName
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
            bindingsMap.put(binding.getExchangeName(),bindingMap);
        }
        //4. 加载消息数据 - queueMessageMap , messageMap
        //查询所有的队列名字,然后根据队列名加载所有的消息
        for (MSGQueue queue : msgQueues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }
        //5. 对于未被确认的消息,如果消费者把消息取走了,但是还没有回复ACK,这段期间如果重启了,这个时候就会断开连接
        // 我们将其设定为如果出现上述场景,我们就认为这个消息消费者还没有取走,我们把消息恢复回队列中,就会让消费者再去处理
        // 这个消息在存储到硬盘上的时候,我们就认为这个消息没有被取走
    }
}